package polimi.disttest;

import java.net.ConnectException;
import java.net.MalformedURLException;
import java.util.HashMap;
import java.util.Map;

import polimi.reds.broker.overlay.AlreadyNeighborException;
import polimi.reds.broker.overlay.GenericOverlay;
import polimi.reds.broker.overlay.NotRunningException;
import polimi.reds.broker.overlay.Overlay;
import polimi.reds.broker.overlay.SimpleTopologyManager;
import polimi.reds.broker.overlay.TCPTransport;
import polimi.reds.broker.overlay.TopologyManager;
import polimi.reds.broker.overlay.Transport;
import polimi.reds.broker.routing.DeferredUnsubscriptionReconfigurator;
import polimi.reds.broker.routing.GenericRouter;
import polimi.reds.broker.routing.GenericTable;
import polimi.reds.broker.routing.HashReplyTable;
import polimi.reds.broker.routing.ImmediateForwardReplyManager;
import polimi.reds.broker.routing.Reconfigurator;
import polimi.reds.broker.routing.ReplyManager;
import polimi.reds.broker.routing.ReplyTable;
import polimi.reds.broker.routing.Router;
import polimi.reds.broker.routing.RoutingStrategy;
import polimi.reds.broker.routing.SubscriptionForwardingRoutingStrategy;
import polimi.reds.broker.routing.SubscriptionTable;
import polimi.reds.optimizer.OptimizerStrategy;
import polimi.reds.optimizer.OptimizerStrategyOptions;
import polimi.reds.optimizer.ReconfiguratorOptimizerDecorator;
import polimi.reds.optimizer.SwapCapableGenericOverlay;
import polimi.reds.optimizer.TrafficLock;

public class Broker {

	private OptimizerStrategyOptions o;
	private Overlay overlay;
	private OptimizerStrategy optimizer;
	
	public Broker(OptimizerStrategyOptions o) {
		this.o=o;
		
		Transport transport = null;
		transport = new TCPTransport(o.getIntegerOption(LocalManager.OPTION_PORT));
		TopologyManager topolMgr = new SimpleTopologyManager();
		TrafficLock trafficLock=new TrafficLock();
		SubscriptionTable subscriptionTable = new GenericTable();
		overlay = new SwapCapableGenericOverlay(trafficLock, topolMgr, transport, subscriptionTable);
		RoutingStrategy routingStrategy = new SubscriptionForwardingRoutingStrategy();
		DeferredUnsubscriptionReconfigurator _reconf = new DeferredUnsubscriptionReconfigurator(false,0);
		Reconfigurator reconf = new ReconfiguratorOptimizerDecorator(_reconf,_reconf,trafficLock);
		ReplyManager replyMgr = new ImmediateForwardReplyManager();
		ReplyTable replyTbl = new HashReplyTable();
		Router router = new GenericRouter(overlay,subscriptionTable,routingStrategy,replyTbl,replyMgr);
		reconf.setRouter(router);
		overlay.addNeighborhoodChangeListener(reconf);
		replyMgr.setRouter(router);
		overlay.getNodeDescriptor().addUrl("reds-tcp:"+o.getOption(LocalManager.OPTION_IP)+":"+o.getIntegerOption(LocalManager.OPTION_PORT));
		try {
			optimizer = (OptimizerStrategy) Class.forName("polimi.reds.optimizer.algorithm."+o.getOption(LocalManager.OPTION_ALGORITHM)).newInstance();
		} catch (Exception e) {
			System.err.println("Unable to find algorithm: " + o.getOption(LocalManager.OPTION_ALGORITHM) + " (" + e.getMessage() + ")");
			System.exit(0);
		}
		optimizer.initialize(overlay, router, convertParams((String) o.getOption(LocalManager.OPTION_ALGORITHM_PARAMETERS)), trafficLock);
		
	}

	public void createOverlay() {
		overlay.start();
	}
	
	public void destroyOverlay() {
		overlay.stop();
	}

	public void connectToNeighbors() throws AlreadyNeighborException, ConnectException, MalformedURLException, NotRunningException {
		String neighborURLsWithCommas = (String) o.getOption(LocalManager.OPTION_INITIAL_NEIGHBORS);
		
		if (neighborURLsWithCommas==null || neighborURLsWithCommas.equals("")) {
			return;
		}
		
		String[] neighborURLs = neighborURLsWithCommas.split(",");
		
		for (String url:neighborURLs) {
			System.out.print("Connecting to '" + url + "'... ");
			overlay.addNeighbor(url);
			System.out.println("done!");
		}
	}

	public Map<String,Object> getStats() {
		
		Map<String,Object> stats = new HashMap<String,Object>();
		stats.putAll(o.createOptionsMap());
		stats.putAll(optimizer.getStats());
		
		return stats;
	}
	
	private Map<String,String> convertParams(String flatParams) {
		HashMap<String,String> map=new HashMap<String,String>();
		String params[]=flatParams.split(",");
		for (String param:params) {
			String nameValue[]=param.split("=");
			if (nameValue.length>1) {
				map.put(nameValue[0], nameValue[1]);
			}
		}
		return map;
	}

}
